package com.atguigu.app

import com.alibaba.fastjson.JSON
import com.atguigu.beans.{OrderDetail, OrderInfo, SaleDetail, UserInfo}
import com.atguigu.gmall.constances.CommonConstants
import com.atguigu.util.MykafkaUtil.MyKafkaUtil
import com.atguigu.util.{MyEsUtil, PropertiesUtil}
import com.google.gson.Gson
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import java.time.format.DateTimeFormatter
import java.time.{LocalDate, LocalDateTime}
import java.util.Properties
import scala.collection.mutable.ArrayBuffer

object OrderDetailApp extends BaseApp {
  override var appName: String = "OrderDetailApp"
  override var internal: Int = 10

  def main(args: Array[String]): Unit = {

    sc=new StreamingContext("local[4]",appName,Seconds(internal))
    sc.sparkContext.setLogLevel("error")
    runApp{
      //ds1  order_detail  ds2  order_info
      val ds1 = MyKafkaUtil.getKafkaStream(CommonConstants.KAFKA_ORDER_DETAIL, sc)
      val ds2 = MyKafkaUtil.getKafkaStream(CommonConstants.KAFKA_ORDER_INFO, sc)

      //1.双流转换为k->v元素形式,需要使用到样例类封装kafka的json串
        //1.1 通过两个样例类封装两个流的json串
        val ds3 = ds1.map(record => {
          val jsonstr = record.value()
          val orderDetail = JSON.parseObject(jsonstr, classOf[OrderDetail])
          (orderDetail.order_id, orderDetail)
        })
      val ds4 = ds2.map(record => {
        val jsonstr = record.value()
        val orderInfo = JSON.parseObject(jsonstr, classOf[OrderInfo])
        val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
        val formatter1 = DateTimeFormatter.ofPattern("yyyy-MM-dd")
        val localDateTime = LocalDateTime.parse(orderInfo.create_time, formatter)
        orderInfo.create_hour = localDateTime.getHour.toString
        val localDate = localDateTime.format(formatter1)
        orderInfo.create_date = localDate;
        //模拟手机号脱敏
        orderInfo.consignee_tel = orderInfo.consignee_tel.replaceAll("(\\w{3})\\w*(\\w{4})", "$1****$2")
        (orderInfo.id,orderInfo)
      })

      val ds5 = ds4.fullOuterJoin(ds3)
      val ds6 = ds5.mapPartitions(partition => {

        //以分区为单位获取redis连接
        val config: Properties = PropertiesUtil.load("config.properties")
        val client: Jedis = new Jedis(config.getProperty("redis.host"), config.getProperty("redis.port").toInt)
        val arrayBuffer = new ArrayBuffer[SaleDetail]
        val gson = new Gson()
        partition.foreach {
          case (order_id, (orderinfo, orderdetail)) => {
            //写入
            if (orderinfo != None) {
              val oiobject = orderinfo.get
              // scala的样例类，使用fastJson转换很麻烦，建议使用gson

              if (orderdetail != None) {
                val odobject = orderdetail.get
                arrayBuffer.append(new SaleDetail(oiobject, odobject))
              }
              client.setex("order_info:" + order_id, 5 * 60 * 2, gson.toJson(oiobject))
              //判断orderDetail是否不为None，不为None，进行匹配
              //到redis中通过key找有没有匹配上的order_detail的,如果没有该key,set会是个空集合
              val preset = client.smembers("order_detail:" + order_id)
              preset.forEach(orderdetailJsonStr => {
                arrayBuffer.append(new SaleDetail(oiobject, gson.fromJson(orderdetailJsonStr, classOf[OrderDetail])))
              })
            } else {
              val odobject = orderdetail.get
              //从redis通过key查找字符串,如果没有该key,则返回null
              val str = client.get("order_info:" + odobject.order_id)
              if (str != null) {
                arrayBuffer.append(new SaleDetail(gson.fromJson(str, classOf[OrderInfo]), odobject))
              } else {
                client.sadd("order_detail:" + order_id, gson.toJson(odobject))
                //设置加入redis的k->v对的过期时间
                client.expire("order_detail:" + order_id, 5 * 60 * 2)
              }
            }

          }
        }
        client.close()
        arrayBuffer.iterator
      })

      // 获取redis的userinfo数据进行关联
      val ds7 = ds6.mapPartitions(partition => {
        val gson = new Gson
        val config: Properties = PropertiesUtil.load("config.properties")
        val client: Jedis = new Jedis(config.getProperty("redis.host"), config.getProperty("redis.port").toInt)
        val details = partition.map(saleDetail => {
          val str: String = client.get("user_info:" + saleDetail.user_id)

          val userInfo = JSON.parseObject(str, classOf[UserInfo])
          saleDetail.mergeUserInfo(userInfo)
          saleDetail
        })
        client.close()
        details
      })

      // 将数据导入es
          //es创建索引
          val ds8 = ds7.map(saleDetail => (saleDetail.order_detail_id, saleDetail))
        ds8.foreachRDD(rdd=>{
          rdd.foreachPartition(partition=>{
            MyEsUtil.insertBulk("gmall2020_sale_detail" +LocalDate.now(),partition.toList )
          })
        })
    }
  }
}
